package mongodb.spark.examples;

import static java.util.Collections.singletonList;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.bson.Document;

import com.mongodb.spark.MongoSpark;
import com.mongodb.spark.config.ReadConfig;
import com.mongodb.spark.rdd.api.java.JavaMongoRDD;

public final class AggregationTest {

	public static void main(final String[] args) throws InterruptedException {

		String uri = "mongodb://dataviewer:dataviewer2017@10.0.14.24:27017/admin.bond_sentiment_news";
		SparkSession spark = SparkSession.builder().master("local").appName("MongoSparkConnectorIntro")
				.config("spark.mongodb.input.uri", uri).config("spark.mongodb.output.uri", uri).getOrCreate();

		// Create a JavaSparkContext using the SparkSession's SparkContext object
		JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());

		// Create a custom ReadConfig
		Map<String, String> readOverrides = new HashMap<String, String>();
		readOverrides.put("database", "dm_bond");
		readOverrides.put("collection", "bond_sentiment_news");
		readOverrides.put("readPreference.name", "secondaryPreferred");
		ReadConfig readConfig = ReadConfig.create(jsc).withOptions(readOverrides);

		// Load and analyze data from MongoDB
		JavaMongoRDD<Document> rdd = MongoSpark.load(jsc, readConfig);

		/* Start Example: Use aggregation to filter a RDD ***************/
		JavaMongoRDD<Document> aggregatedRdd = rdd
				.withPipeline(singletonList(Document.parse("{ $match: { editUser : 'admin' } }")));
		/* End Example **************************************************/

		long start = getNows();
		// Analyze data from MongoDB
		System.out.println(aggregatedRdd.count());
		System.out.println("init-" + (getNows() - start));
		start = getNows();
		System.out.println(aggregatedRdd.count());
		System.out.println("2-" + (getNows() - start));
		start = getNows();
		System.out.println(aggregatedRdd.count());
		System.out.println("3-" + (getNows() - start));

		start = getNows();
		Thread.sleep(1000);
		System.out.println("3=====" + (getNows() - start));

		start = getNows();
		System.out.println(aggregatedRdd.first().toJson());
		System.out.println("4-" + (getNows() - start));

		jsc.close();

	}

	public static long getNows() {
		return new Date().getTime();
	}
}